package com.ehotting.mqbd.core.mq.batch;

import com.ehotting.mqbd.api.enums.DirectionEnum;
import com.ehotting.mqbd.core.helper.MQStringHelper;
import com.ehotting.mqbd.core.mq.consumer.MessageConsumer;
import com.ehotting.mqbd.core.mq.consumer.action.ConsumerActionAble;

import java.util.ArrayList;
import java.util.List;

public class BatchMessageResponseConsumer {

    private List<MessageConsumer> list = new ArrayList();

    public BatchMessageResponseConsumer(MessageConsumer.MessageConsumerParam messageConsumerParam,
                                        BatchConsumerResponseAction batchConsumerRequestAction){

        for(int i=0;i<batchConsumerRequestAction.getList().size();i++){

            ConsumerActionAble consumerAction = batchConsumerRequestAction.getList().get(i);
            String consumerGroup = MQStringHelper.createConsumerGroupAtConsumerApp(
                    consumerAction.getTopic(),
                    consumerAction.getMqApplicationContext(),
                    DirectionEnum.RESPONSE,
                    null);
            MessageConsumer messageConsumer = new MessageConsumer(
                    messageConsumerParam,
                    consumerGroup,
                    consumerAction);
            list.add(messageConsumer);

        }

    }

    public void start(){
        list.stream().forEach(t->{t.start();});
    }

    public void shutdown(){
        list.stream().forEach(t->t.shutdown());
    }

}
